package com.wh.wisdomsite.xbox.core.threadpool;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


/**
 * Job任务队列(接口)
 * @Package com.wh.wisdomsite.xbox.core.threadpool
 * @author 谢泽鹏
 * @date 2014年10月8日 下午4:02:14
 * @Copyright 个人版权所有
 * @Company 贵州万华科技有限公司Copyright (c) 2014
 * @version V1.0
 */
public class JobQueue {

    private static final Log logger = LogFactory.getLog(JobQueue.class);

    //最大线程数
    private int maxQueueSize;

    //JobDispathcher对象
    private JobDispathcher jobDispathcher;

    //
    private String key;

    private BlockingQueue<Job> jobQueue;

    private AtomicInteger counter;

    private ReentrantLock lock;
    /**
     * 是否有资源的锁
     */
    private Condition hasResource;
    /**
     * 是否有任务.避免任务线程空转
     */
    private Condition hasJob;

    private QueueChecker queueChecker;


    /**
     * 构造函数
     * @param key
     * @param maxQueueSize 最大线程数
     * @param jobDispathcher
     */
    public JobQueue(String key, int maxQueueSize, JobDispathcher jobDispathcher) {
        this.maxQueueSize = maxQueueSize;
        this.key = key;
        this.jobDispathcher = jobDispathcher;
        this.jobQueue = new LinkedBlockingDeque<Job>(this.maxQueueSize);
        this.counter = new AtomicInteger(0);

        lock = new ReentrantLock();
        hasResource = lock.newCondition();
        hasJob = lock.newCondition();

        queueChecker = new QueueChecker();
        queueChecker.start();
    }

    public boolean isCompleted() {
        return jobQueue.size() == 0;
    }

    public int size() {
        return this.jobQueue.size();
    }

    public boolean offer(Job job) {
        boolean result = jobQueue.offer(job);
        if (result) {
            counter.incrementAndGet();
            notifyHasJob();
        }
        return result;
    }

    public void clean() {
        counter.set(0);
        jobQueue.clear();
        queueChecker.stopThread();
        queueChecker = null;

    }

    public void blockNoJob(long waitTime) {
        boolean flag = lock.tryLock();
        if (flag) {
            try {
                hasJob.await(waitTime, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                //do nothing
            } finally {
                lock.unlock();
            }
        }
    }

    public void notifyHasJob() {
        boolean flag = false;
        try {
            flag = lock.tryLock(50, TimeUnit.MILLISECONDS);
            if (flag) {
                hasJob.signalAll();
            }
        } catch (InterruptedException e) {
            // do nothing
        } finally {
            if (flag) {
                lock.unlock();
            }
        }
    }

    public void blockNoResource(long waitTime) {

        boolean flag = lock.tryLock();
        if (flag) {
            try {
                hasResource.await(waitTime, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                //do nothing
            } finally {
                lock.unlock();
            }
        }
    }

    public void notifyHasResource() {
        boolean flag = false;
        try {
            flag = lock.tryLock(50, TimeUnit.MILLISECONDS);
            if (flag) {
                hasResource.signalAll();
            }
        } catch (InterruptedException e) {
            //do nothing
        } finally {
            if (flag) {
                lock.unlock();
            }
        }
    }

    class QueueChecker extends Thread {
        boolean isRunning = true;

        public QueueChecker() {
            super("queueChecker-" + key);
        }

        @Override
        public void run() {
            while (isRunning) {

                try {
                    Job job = jobQueue.peek();
                    if (job != null) {
                        if (jobDispathcher.checkResource(job)) {
                            jobQueue.poll();
                            jobDispathcher.inncExecute(job);
                            counter.decrementAndGet();
                        } else {
                            blockNoResource(5000);
                            System.out.println(Thread.currentThread() + "blocNoResource 释放" + jobQueue.size());
                        }
                    } else {
                        blockNoJob(5000);
//                      System.out.println(Thread.currentThread()+"blockNoJob 释放"+jobQueue.size());
                    }
                } catch (Exception e) {
                    logger.error("QueueChecker error", e);
                }
            }
        }

        public void stopThread() {
            try {
                isRunning = false;
                this.interrupt();
            } catch (Exception e) {
                //do nothing
            }
        }
    }

}
